package com.hujf.flink.window.TimeWindow;

import com.hujf.flink.ch05.MySource;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author hujf
 * @Date 2022/5/23 16:37
 * @Description TODO
 */
public class AggregateTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Tuple2<String,Long>> dataStreamSource = environment.fromElements(
                Tuple2.of("BWM", 2L),
                Tuple2.of("BWM", 4L),
                Tuple2.of("tESLA", 2L),
                Tuple2.of("tESLA", 9L),
                Tuple2.of("tESLA", 3L),
                Tuple2.of("tESLA", 2L)

        );

        SingleOutputStreamOperator<Double> aggregate = dataStreamSource.keyBy(0)
                .countWindow(2)
                .aggregate(new AggregateFunctionss());

        aggregate.print();
        environment.execute();

    }
    
    private static class AggregateFunctionss implements AggregateFunction< Tuple2<String,Long>, Tuple2<Long,Long>,Double>{

        @Override
        public Tuple2<Long, Long> createAccumulator() {
            return new Tuple2<>(0L,0L);
        }

        @Override
        public Tuple2<Long, Long> add(Tuple2<String, Long> value, Tuple2<Long, Long> accumulator) {
            return new Tuple2<>(accumulator.f0+value.f1,accumulator.f1+1L);
        }

        @Override
        public Double getResult(Tuple2<Long, Long> accumulator) {
            return ((double)accumulator.f0)/accumulator.f1;
        }

        @Override
        public Tuple2<Long, Long> merge(Tuple2<Long, Long> a, Tuple2<Long, Long> b) {
            return new Tuple2<>(a.f0+b.f0,a.f1+b.f1);
        }
    }
}
